package com.wyx.springbootspringkafkademo;

import com.wyx.springbootspringkafkademo.producer.KafkaProducerService;
import jakarta.annotation.PostConstruct;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Service;

import java.util.Collections;

@Service
public class KafkaConsumerOffsetService {

    private final ConsumerFactory<String, String> consumerFactory;

    public KafkaConsumerOffsetService(ConsumerFactory<String, String> consumerFactory) {
        this.consumerFactory = consumerFactory;
    }

    @PostConstruct
    public void init() {
        try (Consumer<String, String> consumer = consumerFactory.createConsumer()) {
            consumer.subscribe(Collections.singletonList(KafkaProducerService.TOPIC));

            // 等待消费者分配到分区（这里只是为了示例，实际使用中可能需要更复杂的逻辑）
            consumer.poll(java.time.Duration.ofMillis(100));

//            Map<TopicPartition, Long> assignedPartitions = consumer.assignment().stream()
//                .collect(Collectors.toMap(partition -> partition, partition -> null)); // 初始化为 null，实际需要获取已提交偏移量
//
//            assignedPartitions.forEach((partition, offset) -> {
//                // 这里可以设置偏移量，例如从已提交的偏移量开始消费
//                System.out.println(partition.topic() + "-" + partition.partition()+offset);
//            });
            // 获取已提交的偏移量（这里需要结合 KafkaConsumer 的 committed 方法）
            // 注意：Spring Kafka 的 ConsumerFactory 创建的 Consumer 是轻量级的，不直接暴露 committed 方法
            // 因此，你可能需要直接使用 KafkaConsumer 或通过其他方式获取已提交偏移量

            // 示例：手动获取已提交偏移量（假设你有一个 KafkaConsumer 实例）
//            KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties); // 替换为你的配置
//            kafkaConsumer.assign(assignedPartitions.keySet());
//            for (TopicPartition partition : assignedPartitions.keySet()) {
//                OffsetAndMetadata committed = kafkaConsumer.committed(partition);
//                if (committed != null) {
//                    System.out.println("Partition: " + partition.partition() + ", Committed Offset: " + committed.offset());
//                }
//            }
//            kafkaConsumer.close();

            // 由于 Spring Kafka 的 Consumer 是线程安全的且由 Spring 管理，通常不需要手动创建 KafkaConsumer 实例
            // 你可以通过其他方式（如 KafkaAdmin 或自定义逻辑）来获取已提交偏移量
        }
    }
}